package com.hxzy.websocket;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.log4j.Log4j2;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * @author Administrator
 */
@Component
@Log4j2
// 接口路径 ws://ip:9380/webSocket/userId;
@ServerEndpoint("/websocket/{userId}")
public class WebSocketService {

    //与某个客户端的连接会话，需要通过它来给客户端发送数据 (tcp长链接)
    private Session session;

    /**
     * 用户ID（全局唯一的）
     */
    private String userId;

    /**
     * concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
     * 虽然@Component默认是单例模式的，但springboot还是会为每个websocket连接初始化一个bean，所以可以用一个静态set保存起来。
     */
    private static Set<WebSocketService> webSockets =new CopyOnWriteArraySet<>();

    /**
     * 用来存在线连接用户信息
     */
    private static ConcurrentHashMap<String,Session> sessionPool = new ConcurrentHashMap<String,Session>();

    /**
     * 1、链接成功调用的方法(用户第一次打开连接) ( 前端建立连接按钮)
     */
    @OnOpen
    public void onOpen(Session session, @PathParam(value="userId")String userId) {
        try {
            this.session = session;
            this.userId = userId;
            webSockets.add(this);
            sessionPool.put(userId, session);
            log.info("【websocket消息】有新的连接，总数为:"+webSockets.size());

            String msg=buildMessage("系统管理员","欢迎光临");
            this.session.getAsyncRemote().sendText(msg);

            //向所有在线用户发送一个刷新在用户的列表消息
            String onLine=buildOneLineCount();
            this.sendAllMessage(onLine);

        } catch (Exception e) {
        }
    }

    /**
     * 构建消息
     * @param userId  发送消息人
     * @param message 消息内容
     * @return
     */
    private String buildMessage(String userId,String message) throws JsonProcessingException {
        Map<String,Object> mp=new HashMap<>();
        mp.put("type",2);
        mp.put("userId",userId);
        mp.put("content",message);
        ObjectMapper  mapper=new ObjectMapper();
        String str=mapper.writeValueAsString(mp);
        return str;
    }

    /**
     * 构建消息在线人数
     * @return
     */
    private String buildOneLineCount() throws JsonProcessingException {
        Map<String,Object> mp=new HashMap<>();
        mp.put("type",1);
        mp.put("count",webSockets.size());
        mp.put("userList",sessionPool.keys());
        ObjectMapper  mapper=new ObjectMapper();
        String str=mapper.writeValueAsString(mp);
        return str;
    }

    /**
     * 2、链接关闭调用的方法(正常退出，前端关闭连接按钮)
     */
    @OnClose
    public void onClose() {
        try {
            webSockets.remove(this);
            sessionPool.remove(this.userId);
            log.info("【websocket消息】连接断开，总数为:"+webSockets.size());

            //向所有在线用户发送一个刷新在用户的列表消息
            String onLine=buildOneLineCount();
            this.sendAllMessage(onLine);

        } catch (Exception e) {
        }
    }
    /**
     * 3、 收到客户端消息后调用的方法
     *
     * @param message
     * @param message
     */
    @OnMessage
    public void onMessage(String message) {
        log.info("【websocket消息】收到客户端消息:"+message);
        ObjectMapper objectMapper=new ObjectMapper();
        try {
            Map map = objectMapper.readValue(message, Map.class);

            //收到的内容
            //{'from':this.userId, 'to': this.selectUser, 'content': this.contentText};
            String from= map.get("from").toString();
            String receiveUserId = map.get("to").toString();
            String content=map.get("content").toString();
            // 向所有人发消息
            if(receiveUserId.equals("所有人")){
                String msg=this.buildMessage(from,content);
                this.sendAllMessage(msg);
            }else{
                //向指定的用户发送消息
                this.sendOneMessage(from,receiveUserId,content);
            }

        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }

    }

    /** 4、异常 发送错误时的处理
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("用户错误,原因:"+error.getMessage());
        error.printStackTrace();
    }


    /**
     * 此为广播消息
     * @param message
     */
    public void sendAllMessage(String message) {
        log.info("【websocket消息】广播消息:"+message);
        for(WebSocketService webSocket : webSockets) {
            try {
                if(webSocket.session.isOpen()) {
                   //向某一个客户发送消息
                    webSocket.session.getAsyncRemote().sendText(message);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 此为单点消息
     * @param from 发送者
     * @param to   接收者
     * @param message
     */
    public void sendOneMessage(String from,String to, String message) {
        Session session = sessionPool.get(to);
        if (session != null && session.isOpen()) {
            try {
                log.info("【websocket消息】 单点消息:"+message);
                String msg=this.buildMessage(from,message);
                session.getAsyncRemote().sendText(msg);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 此为单点消息(多人)
     * @param from 发送者
     * @param userIds
     * @param message
     */
    public void sendMoreMessage(String from,String[] userIds, String message) {
        for(String userId:userIds) {
            Session session = sessionPool.get(userId);
            if (session != null&&session.isOpen()) {
                try {
                    log.info("【websocket消息】 单点消息:"+message);
                    String msg=this.buildMessage(from,message);
                    session.getAsyncRemote().sendText(msg);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }

    }

}